package com.boot.study.api

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import scala.util.Random


// 定义样例类，温度传感器
case class SensorReading(id: String, timeStamp: Long, temperature: Double)

object SourceTest {
  def main(args: Array[String]): Unit = {
    // 创建执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //    // 从集合中读取数据
    //    val dataList: List[SensorReading] = List(
    //      SensorReading("Sensor_1", 1547718759, 35.8),
    //      SensorReading("Sensor_3", 1547818759, 20.9),
    //      SensorReading("Sensor_5", 1547719759, 15.2),
    //      SensorReading("Sensor_9", 1547718769, 32.8)
    //    )
    //    val stream: DataStream[SensorReading] = env.fromCollection(dataList)
    //    stream.print()

    //    // 读取文件
    //    val stream1: DataStream[String] = env.readTextFile("D:\\WorkSpace\\idea\\Flink\\src\\main\\resources\\sensor.txt")
    //    stream1.print()

    // 从kafka中读取数据
    // kafka执行命令 kafka-console-producer.bat --broker-list 127.0.0.1:9092 --topic sensor
    //    val properties = new Properties()
    //    properties.setProperty("bootstrap.servers", "127.0.0.1:9092")
    //    properties.setProperty("group.id", "consumer-group")
    //    val stream3: DataStream[String] = env.addSource(new FlinkKafkaConsumer011[String]("sensor", new SimpleStringSchema(), properties))
    //    stream3.print()

    // 自定义source
    val stream4: DataStream[SensorReading] = env.addSource(new MySensorSource())
    stream4.print()
    // 执行
    env.execute("source test")
  }
}

// 自定义source
class MySensorSource extends SourceFunction[SensorReading] {

  // 定义标识位来表示数据源是否正常发送数据
  var running: Boolean = true

  override def run(ctx: SourceFunction.SourceContext[SensorReading]): Unit = {
    // 定义随机数发生器
    val rand = new Random()
    // 随机生成一组（10）个传感器数据 （id,温度）
    var curTemp = 1.to(10).map(i => ("Sensor_" + i, rand.nextDouble() * 10))
    // 定义无限循环不停的发送数据，除非被cancel
    while (running) {
      // 获取当前时间搓
      val curTime = System.currentTimeMillis()
      // 在上次数据基础上更新温度值
      curTemp.foreach(data =>
        ctx.collect(SensorReading(data._1, curTime, data._2 + rand.nextGaussian()))
      )
      curTemp = curTemp.map(data => (data._1, data._2 + rand.nextGaussian()))
      // 间隔时间
      Thread.sleep(5000)
    }
  }

  override def cancel(): Unit = running = false
}
